package test

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object hh {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext

    val student1:RDD[String] = sc.textFile("src/main/resources/student.txt")
    student1.foreach(println)
    val studentSet:Dataset[String] = spark.read.textFile("src/main/resources/student.txt")
    studentSet.printSchema()
    studentSet.show()
    //结构化数据,封装数据映射成表结构的元数据
    val studentFrame:DataFrame = spark.read.csv("src/main/resources/student.txt")
    studentFrame.printSchema()
    studentFrame.show()
    studentFrame.createTempView("score")

    val res:DataFrame =spark.sql(
      """
        |select
        | _c2 as term,
        | avg(_c3) as avg_score,
        | max(_c3) as max_score,
        | sum(_c3) as su_score
        | from score
        | group by _c2
        |""".stripMargin
    )

    val res1 = spark.sql(
      """
        |select
        |_c1
        |from score
        |where _c3 >85
        |""".stripMargin
    )
    val data2:DataFrame = spark.read
      .option("encoding","GBK")
      .option("header","true")
      .csv("src/main/resources/23data2.csv")
    data2.printSchema()
    data2.show(60)

    //练习：求平时成绩的平均分、最高分和不及格人数

        // 读取 CSV 文件
        val df = spark.read
          .option("header", "true")
          .option("inferSchema", "true")
          .csv("src/main/resources/23data2.csv")

       df.createOrReplaceTempView("students")
        // 计算平时成绩的平均分
        val averageGrade = spark.sql("SELECT AVG('平均成绩') FROM students").collect()(0).getDouble(0)
        // 计算平时成绩的最高分
        val maxGrade = spark.sql("SELECT MAX('平均成绩') FROM students").collect()(0).getDouble(0)
        // 筛选出不及格的学生
        val failedStudents = spark.sql("SELECT name FROM students WHERE平均成绩 <60")
        val failedCount = failedStudents.count()
        val failedNames = failedStudents.select("name").collect().map(_.getString(0)).toList
        // 输出结果
        println(s"平时成绩平均分：$averageGrade")
        println(s"平时成绩最高分：$maxGrade")
        println(s"不及格人数：$failedCount")
        println(s"不及格学生姓名：$failedNames")
        // 停止 SparkSession
        spark.stop()
      }

}
